package com.yq.canalkafka.service.impl;

import com.alibaba.otter.canal.client.kafka.protocol.KafkaMessage;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.google.common.base.Strings;
import com.yq.canalkafka.enums.MethodEnum;
import com.yq.canalkafka.mapper.CommonMapper;
import com.yq.canalkafka.service.CanalDealService;
import com.yq.canalkafka.service.CanalSqlService;
import com.yq.canalkafka.util.CanalKafkaUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import java.util.List;

/**
 * @author by shizan
 * @Classname CanalDealServiceImpl
 * @Description 处理service实现
 * @Date 2020/11/23 4:28 下午
 */
@Service
@Slf4j
public class CanalDealServiceImpl implements CanalDealService {
    @Autowired
    private CanalSqlService canalSqlService;
    @Autowired
    private CommonMapper commonMapper;

    @Override
    public void printEntries(KafkaMessage message) {
        for (CanalEntry.Entry entry : message.getEntries()) {
            try {
                if (entry.getEntryType() != CanalEntry.EntryType.ROWDATA) {
                    continue;
                }
                String tableName = entry.getHeader().getTableName();
                String databaseName = entry.getHeader().getSchemaName();


                CanalEntry.RowChange rowChange = CanalEntry.RowChange.parseFrom(entry.getStoreValue());

                log.info(String.format("================> binlog[%s:%s] , name[%s,%s] , eventType : %s",
                        entry.getHeader().getLogfileName(), entry.getHeader().getLogfileOffset(),
                        entry.getHeader().getSchemaName(), entry.getHeader().getTableName(), rowChange.getEventType()));
                dealSql(rowChange, tableName);

            } catch (Exception e) {
                log.error("同步异常 offset->{}", message.getOffset());
                log.error("同步异常", e);

            }
        }
    }

    /**
     * sql组装
     *
     * @param rowChange
     * @param tableName
     */
    private void dealSql(CanalEntry.RowChange rowChange, String tableName) {
        for (CanalEntry.RowData rowData : rowChange.getRowDatasList()) {
            String sql = "";
            try {
                String method = rowChange.getEventType().getValueDescriptor().getName();

                switch (rowChange.getEventType()) {
                    case INSERT:
                        sql = canalSqlService.getInsertSql(rowData.getAfterColumnsList(), tableName);
                        send(sql, method);
                        break;
                    case UPDATE:
                        sql = canalSqlService.getUpdateSql(rowData.getAfterColumnsList(), tableName);
                        send(sql, method);
                        break;
                /*
                case DELETE:
                    System.out.println("DELETE ");
                    printColumns(rowData.getBeforeColumnsList());
                    break;
                    */
                    default:
                        break;
                }
            } catch (Exception e) {
                log.error("同步异常,sql处理异常", e);
                log.error("异常sql->{}", sql);
//                log.error("原始数据如下");
//                CanalKafkaUtils.printColumn(rowData.getAfterColumnsList());
            }

        }
    }

    /**
     * sql语句执行
     *
     * @param sql
     * @param method
     */
    private void send(String sql, String method) {
//        log.info("sql->{}",sql);
        if (!Strings.isNullOrEmpty(sql)) {
            if (MethodEnum.UPDATE.getValue().equals(method)) {
                commonMapper.executeUpdate(sql);
            } else if (MethodEnum.INSERT.getValue().equals(method)) {
                commonMapper.executeInsert(sql);
            } else {
                log.error("no execute method :: {} sql::{}", method, sql);
            }
        }
    }
}
